/*
 *  policy - Part of the Policy Server Application
 *  Copyright (C) 2009 Liam Lowey
 *  
 *      http://policyserver.turnitin.co.uk/
 *
 *  Licensed under the Apache License, Version 2.0 (the "License"); 
 *  you may not use this file except in compliance with the License. 
 *  You may obtain a copy of the License at 
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0 
 *  
 *  Unless required by applicable law or agreed to in writing, software 
 *  distributed under the License is distributed on an "AS IS" BASIS, 
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 *  See the License for the specific language governing permissions and 
 *  limitations under the License. 
 */

using System;
using System.Collections.Generic;
using System.Threading;
using System.Net.Sockets;

namespace policyserver
{
  class Reactor
  {
    private static readonly int SELECT_WAIT = (int)(60 * TimeMath.MICROS_IN_A_SEC);

    private SocketCommandQueue _cmdQueue = new SocketCommandQueue();
    private Thread _thread;
    private Dictionary<Socket, Connection> _connectionTable = new Dictionary<Socket,Connection>();
    private List<Connection> _connectionList;
    private List<Socket> _listeningSockets;
    private bool _running = true;
    private int _id;


    public Reactor( int id )
    {
      _id = id;

      _connectionTable.Add( _cmdQueue.socket, null );

      _thread = new Thread( this.run );

      _thread.Start();
    }
    

    public void addConnection( Connection con )
    {
      _cmdQueue.push( Factory.makeCommand( this.doAddConnection, con ) );
    }


    public void stop()
    {
      _cmdQueue.pushUrgent( Factory.makeCommand( this.doStop ) );
      _thread.Join();
    }


    private void run()
    {
      ICommand cmd = null;

      while( _running )
      {
        if( _listeningSockets != null && _listeningSockets.Count > 0 )
        {
          List<Socket> listeners = new List<Socket>( _listeningSockets );
          List<Socket> errorSocks = new List<Socket>( _listeningSockets );
          List<Socket> writers = buildWritableList();

          Socket.Select( listeners, writers, errorSocks, SELECT_WAIT );

          bool updateSocketTable = false;

          //  Process writer sockets
          foreach( Socket sock in writers )
          {
            if( !processWriteSocket( sock ) )
            {
              updateSocketTable = true;
            }
          }

          //  Process reader sockets
          foreach( Socket sock in listeners )
          {
            if( !processReadSocket( sock ) )
            {
              updateSocketTable = true;
            }
          }

          //  Process sockets in error
          foreach( Socket sock in errorSocks )
          {
            _connectionTable.Remove( sock );
            updateSocketTable = true;
          }


          //  Rebuild socket table if necessary
          if( updateSocketTable )
          {
            rebuildSocketList();
          }

          //  process the command queue.
          do
          {
            cmd = _cmdQueue.pop( TimeMath.spanFromMillis( 1 ) );

            if( cmd != null )
            {
              cmd.execute();
            }
          }
          while( _cmdQueue.hasCommand );
        }
        else
        {
          do
          {
            cmd = _cmdQueue.pop( TimeSpan.FromSeconds(3600) );
            if( cmd != null )
            {
              cmd.execute();
            }
          }
          while( _cmdQueue.hasCommand );
        }
      }

      trace( "Reactor {0} exiting.", _id );
    }


    private void doStop()
    {
      _running = false;
    }


    private void doAddConnection( Connection con )
    {
      _connectionTable.Add( con.socket, con );
      rebuildSocketList();
    }


    private void rebuildSocketList()
    {
      _listeningSockets = new List<Socket>( _connectionTable.Count );

      _connectionList = new List<Connection>( _connectionTable.Count );

      if( _connectionTable.Count > 0 )
      {
        foreach( KeyValuePair<Socket,Connection> pair in _connectionTable )
        {
          _listeningSockets.Add( pair.Key );
          _connectionList.Add( pair.Value );
        }
      }
    }


    private List<Socket> buildWritableList()
    {
      List<Socket> writers = new List<Socket>();

      foreach( Connection conn in _connectionList )
      {
        if( null != conn && conn.isWritable )
        {
          writers.Add( conn.socket );
        }
      }

      return writers;
    }


    private bool processReadSocket( Socket sock )
    {
      try
      {
        Connection conn = _connectionTable[sock];
        if( null != conn )
        {
          if( !conn.processRead() )
          {
            _connectionTable.Remove( sock );
            return false;
          }
        }
      }
      catch( SocketException )
      {
        _connectionTable.Remove( sock );
        return false;
      }
      return true;
    }


    private bool processWriteSocket( Socket sock )
    {
      try
      {
        Connection conn = _connectionTable[sock];
        if( null != conn )
        {
          if( !conn.processWrite() )
          {
            _connectionTable.Remove( sock );
            return false;
          }
        }
      }
      catch( SocketException )
      {
        _connectionTable.Remove( sock );
        return false;
      }
      return true;
    }


    private static void trace( string format, params object[] args )
    {
      System.Console.WriteLine( format, args );
    }
  }
}
